package com.lyon.custom.core;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.lang.TypeReference;
import cn.hutool.core.text.CharSequenceUtil;
import cn.hutool.core.thread.NamedThreadFactory;
import cn.hutool.crypto.digest.DigestUtil;
import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpStatus;
import cn.hutool.json.JSONUtil;
import com.alibaba.nacos.client.utils.ParamUtil;
import com.alibaba.nacos.common.utils.UuidUtils;
import com.google.common.collect.Maps;
import com.lyon.custom.api.model.CommonResult;
import com.lyon.custom.api.model.ConfigBaseReqVO;
import com.lyon.custom.api.model.ConfigListenerReqVO;
import com.lyon.custom.common.agent.HttpAgent;
import com.lyon.custom.common.constants.Null;
import com.lyon.custom.common.core.GroupKeys;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

/**
 * @author LeeYan9
 * @since 2021-12-01
 */
@Slf4j
public class ClientWorker {


    private static AtomicReference<Map<String, CacheData>> cacheMap = new AtomicReference<Map<String, CacheData>>(Maps.newConcurrentMap());
    private String tenantId;
    private HttpAgent agent;
    private String serverAddr;
    private static final String CONFIG_URI = "/v1/config/server";
    private volatile int currentTaskId = 0;
    private ScheduledExecutorService executor;
    private ScheduledExecutorService taskExecutor;

    public ClientWorker(Properties properties, HttpAgent agent) {
        // 环境
        this.tenantId = properties.getProperty("namespace");
        this.serverAddr = properties.getProperty("serverAddr");
        this.agent = agent;
        // 轮询客户端调度线程
        executor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("polling-Client-", true));
        // 轮询客户端 工作线程池
        taskExecutor = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(),
                new NamedThreadFactory("polling-Request-Worker", true)
        );

        // 每隔10ms 检查是否有新配置 , 进行分片拉取
        executor.scheduleWithFixedDelay(() -> {
            try {
                checkConfigItemChange();
            } catch (Throwable e) {
                e.printStackTrace();
            }
        }, 1, 10, TimeUnit.MILLISECONDS);
    }

    public String[] getConfigServer(String group, String dataId, long timeout) {
        return getConfigServer(tenantId, group, dataId, timeout);
    }

    public String[] getConfigServer(String tenantId, String group, String dataId, long timeout) {
        ConfigBaseReqVO baseReqVO = ConfigBaseReqVO.builder()
                .tenantId(tenantId)
                .dataId(dataId)
                .group(group)
                .build();
        HttpResponse httpResponse = agent.get(CharSequenceUtil.format("{}{}", serverAddr, CONFIG_URI), BeanUtil.beanToMap(baseReqVO), (int) timeout).execute();
        if (httpResponse.getStatus() == HttpStatus.HTTP_OK) {
            TypeReference<CommonResult<String>> typeReference = new TypeReference<>() {
            };
            CommonResult<String> commonResult = JSONUtil.toBean(httpResponse.body(), typeReference, true);
            if (commonResult != null) {
                return new String[]{commonResult.getData(), httpResponse.header("CONFIG-TYPE")};
            }
            return new String[0];
        }
        return new String[0];
    }

    private AtomicInteger count = new AtomicInteger(0);

    public void checkConfigItemChange() {
        // 线程分片执行, 每个线程监听3000条groupKey数据
        int shardCount = (int) Math.ceil(cacheMap.get().size() / ParamUtil.getPerTaskConfigSize());
        if (currentTaskId < shardCount) {
            for (int i = currentTaskId; i < shardCount; i++) {
                log.info("创建第{}个线程去调度执行",count.incrementAndGet());
                taskExecutor.execute(new LongPollingClient(i));
            }
            currentTaskId = shardCount;
        }
    }
    public void addListener(String dataId, String group, Listener listener) {
        String groupKey = GroupKeys.buildGroupKey(tenantId, group, dataId);
        CacheData cacheData = cacheMap.get().get(groupKey);
        if (cacheData == null) {
            synchronized (cacheMap) {
                if ((cacheData = cacheMap.get().get(groupKey)) == null) {
                    cacheData = new CacheData();
                    cacheData.setTenantId(tenantId);
                    cacheData.setGroup(group);
                    cacheData.setDataId(dataId);
                    cacheData.setGroupKey(groupKey);
                    int taskId = cacheMap.get().size() / (int) ParamUtil.getPerTaskConfigSize();
                    cacheData.setTaskId(taskId);
                    cacheMap.get().put(groupKey, cacheData);
                }
            }
        }
        cacheData.addListener(listener);
    }

    class LongPollingClient implements Runnable {

        private int taskId;

        public LongPollingClient(int taskId) {
            this.taskId = taskId;
        }

        private String uid = UuidUtils.generateUuid();
        private AtomicInteger count = new AtomicInteger(0);

        @Override
        public void run() {
            // 只处理当前分片里的数据
            Collection<CacheData> caches = new ArrayList<>();
            for (CacheData cacheData : cacheMap.get().values()) {
                if (taskId == cacheData.getTaskId()) {
                    caches.add(cacheData);
                }
            }

            try {
                List<String> changeKeys = checkChangeKeys(caches);
                // 更新CacheData
                if (CollUtil.isNotEmpty(changeKeys)) {
                    for (String changeKey : changeKeys) {
                        String[] split = changeKey.split(";");
                        String dataId = split[2];
                        String group = split[1];
                        String tenant = split[0];
                        String groupKey = CharSequenceUtil.join(";", tenant, group, dataId);
                        String[] configData = getConfigServer(tenant, group, dataId, 30000);
                        // 记录数据；

                        // 更新缓存
                        CacheData cacheData = cacheMap.get().get(groupKey);
                        cacheData.setContent(configData[0]);
                        cacheData.setType(configData[1]);
                        // 发布变更事件
                        cacheData.checkMd5Change();
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            // 持续调度
            log.info("调度线程[{}]：第[{}]次", uid, count.incrementAndGet());
            taskExecutor.execute(this);
        }

        private List<String> checkChangeKeys(Collection<CacheData> caches) {
            if (CollUtil.isNotEmpty(caches)) {
                int timeout = 10000;
                Map<String, String> clientMd5Map = caches.stream().collect(Collectors.toMap(CacheData::getGroupKey, CacheData::getMd5, (old, now) -> old));
                ConfigListenerReqVO listenerReqVO = ConfigListenerReqVO.builder()
                        .pollingTimeout(timeout)
                        .sync(false)
                        .tenantId(tenantId)
                        .clientMd5Map(clientMd5Map)
                        .build();
                String resultJson = agent.doPost(CharSequenceUtil.format("{}{}/listener", serverAddr, CONFIG_URI), JSONUtil.toJsonStr(listenerReqVO), (int) (timeout + (long) Math.round(timeout >> 1)));
                if (CharSequenceUtil.isBlank(resultJson)) {
                    return ListUtil.empty();
                }
                return JSONUtil.toList(resultJson, String.class);
            }
            return ListUtil.empty();
        }
    }

    @Data
    public static class CacheData {

        // 当前的监听器
        private List<ListenerWrap> listeners = new CopyOnWriteArrayList<>();
        private int taskId;

        void addListener(Listener listener) {
            listeners.add(ListenerWrap.wrap(listener));
        }

        public void setContent(String content) {
            this.content = content;
            this.md5 = DigestUtil.md5Hex(content);
        }

        public void checkMd5Change() {
            for (ListenerWrap listenerWrap : listeners) {
                if (!CharSequenceUtil.equals(md5, listenerWrap.getMd5Content())) {
                    safeNotifyListener(tenantId, group, dataId, type, listenerWrap);
                }
            }
        }

        private void safeNotifyListener(String tenantId, String group, String dataId, String type, ListenerWrap listenerWrap) {
            this.tenantId = tenantId;
            Listener listener = listenerWrap.getListener();
            Runnable runnable = () -> {

                if (listener instanceof AbstractChangeShareListener) {
                    AbstractChangeShareListener shareListener = ((AbstractChangeShareListener) listener);
                    shareListener.fillProp(group, dataId, type);
                }
                listener.receiveConfigInfo(content);

                if (listener.getClass().isAssignableFrom(AbstractConfigChangeListener.class)) {
                    try {
                        Map parseChangeData = ConfigChangeHandler.getInstance().parseChangeData(listenerWrap.getLastContent(), content, type);
                        assert listener instanceof AbstractConfigChangeListener;
                        ((AbstractConfigChangeListener) listener).receiveConfigChange(new ConfigChangeEvent(parseChangeData));
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            };
            try {
                Executor executor = listener.getExecutor();
                if (executor != null) {
                    executor.execute(runnable);
                } else {
                    runnable.run();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        private volatile String tenantId;
        private volatile String group;
        private volatile String dataId;
        private volatile String groupKey;
        private volatile String md5 = Null.NULL;
        private volatile String content;
        private volatile String type;

    }

    @Data
    static class ListenerWrap {
        private Listener listener;
        private String md5Content;
        private String lastContent;

        private ListenerWrap() {
        }

        public static ListenerWrap wrap(Listener listener) {
            ListenerWrap listenerWrap = new ListenerWrap();
            listenerWrap.setListener(listener);
            return listenerWrap;
        }
    }

}
